""" 
Copyright 2009 Texas A&M University
 
Licensed under the Apache License, Version 2.0 (the "License"); you may not use 
this file except in compliance with the License. You may obtain a copy of the 
License at

http://www.apache.org/licenses/LICENSE-2.0
  
Unless required by applicable law or agreed to in writing, software distributed 
under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR 
CONDITIONS OF ANY KIND, either express or implied. See the License for the 
specific language governing permissions and limitations under the License.
"""
from logging import getLogger
from Queue import Queue, Empty
from twisted.internet.defer import Deferred, DeferredList, DeferredLock
from twisted.internet.defer import AlreadyCalledError
from twisted.internet.threads import deferToThread

import zope.interface
try:
    import threading
except ImportError:
    import dummy_threading as threading

from zope.component import getGlobalSiteManager

from meercat.interfaces import IMetadata
from meercat.base.metadata import Metadata
from meercat.mapreduce import interfaces
from meercat.mapreduce.registry import GlobalMapperRegistry
from meercat.mapreduce.exceptions import MapperError, NoPath, UnkownFormat



log = getLogger('meercat.mapreduce.defer')

"""
try:
    import multiprocessing
    multiprocessing.log_to_stderr()
    log = multiprocessing.get_logger()
except ImportError:
    pass 
"""

class MapReduceDeferred(Deferred):
    def __init__(self, input, targetFormat, deferredFactory=deferToThread,
                 errorBackMapErrors=True, errorBackMapPathErrors=False,
                 maxWorkers=None, mapperRegistry=GlobalMapperRegistry,
                 componentRegistry=None,
                 lockFactory=threading.RLock):
        Deferred.__init__(self)
        self._errors = False
        self._mapErrors = bool(errorBackMapErrors)
        self._pathErrors = bool(errorBackMapPathErrors)
        self._targetFormat = targetFormat
        self._factory = deferredFactory
        self._maxWorkers = maxWorkers
        self._mapperRegistry = mapperRegistry
        if componentRegistry is None:
            componentRegistry = getGlobalSiteManager()
        self._components = componentRegistry
        self._reducer = self._components.getUtility(interfaces.IReducer,
                                                    self._targetFormat)
        self._lock = lockFactory()
        self._mapQueue = Queue()
        self._reduceQueue = Queue()
        
        self._runningMaps = 0
        self._runningReduces = 0
        
        self._start_maps(input)
    
    def _start_maps(self, input):
        log.debug('start maps: %s' % (str(len(input))))
        
        for i in input:
            self._mapQueue.put(i)
        
        if self._maxWorkers is not None and self._maxWorkers < len(input):
            for i in range(0,maxWorkers):
                self._dispatch_map(self._mapQueue.get())
        else:
            while not self._mapQueue.empty():
                self._dispatch_map(self._mapQueue.get())
    
    def _dispatch_map(self, metadata):
        self._lock.acquire()
        #log.debug(u'dispatch_map(metadata=%s)' % (metadata.getValue()))
        log.debug(u'dispatch_map()')
        try:
            if self._errors:
                log.error('Error already encountered in this map reduce, no more dispatching')
                return
            
            if not IMetadata.providedBy(metadata):
                self._errors = True
                self.errback('Invalid input, input must implement the '
                             'meercat.interfaces.IMetadata interface')
            
            mapper = self._mapperRegistry.getMapper(metadata.getFormat(),
                                                    self._targetFormat)
            
            d = self._factory(mapper.map,Metadata(format=metadata.getFormat(),
                                                  value=metadata.getValue(),
                                                  source=metadata.getSource()))
            self._runningMaps += 1
            d.addCallbacks(self._map_finished, self._map_failure)
        except (NoPath, UnkownFormat), e:
            if self._pathErrors:
                log.error('Path error: %s' % (str(e)))
                self._errors = True
                self.errback(e)
                return
            else:
                log.debug('No path or unkown format: %s' % (str(e)))
                self._runningMaps += 1
                self._map_finished(None)
                return
        except Exception, e:
            self._errors = True
            raise
            self.errback(e)
            return
            
        finally:
            self._lock.release()
        
        
    def _map_finished(self, metadata):
        self._runningMaps -= 1
        if metadata is not None and metadata.getValue() is not None:
            log.debug(u'map_finished')
            self._reduceQueue.put(metadata)
        try:
            self._dispatch_map(self._mapQueue.get_nowait())
            return
        except Empty:
            log.debug(u'map_finished: empty')
            pass
        
        self._dispatch_reduce()
    
    def _map_failure(self, failure):
        self._runningMaps -= 1
        log.error('Map failure callback')
        if self._mapErrors:
            self._errors = True
            self.errback(failure)
            return
        log.error('Map failure: %s' % (str(failure)))
        self._dispatch_reduce()
    
    def _dispatch_reduce(self):
        self._lock.acquire()
        log.debug('dispatch_reduce')
        try:
            if self._errors:
                return
            one = None
            two = None
            one = self._reduceQueue.get_nowait()
            two = self._reduceQueue.get_nowait()
            
            d = self._factory(self._reducer.reduce,one,two)
            d.addCallbacks(self._reduce_finished,self._reduce_failure)
            self._runningReduces += 1
            return
        except Empty:
            if one is not None:
                log.debug('reduce queue had 1')
                self._reduceQueue.put(one)
            else:
                log.debug('reduce queue was empty')
            
            self._check_finished()
        except Exception, e:
            log.debug('Unexpected exception: %s' % str(e))
            self._errors = True
            self.errback(e)
        finally:
            self._lock.release()
    
    def _check_finished(self):
        self._lock.acquire()
        log.debug('check_finished')
        try:
            if not self._errors and self._mapQueue.empty() and \
               self._runningMaps == 0 and self._runningReduces == 0:
                if self._reduceQueue.qsize() == 1:
                    log.debug('callback with value')
                    self.callback(self._reduceQueue.get())
                elif self._reduceQueue.empty():
                    try:
                        log.debug('callback none')
                        self.callback(None)
                    except AlreadyCalledError:
                        pass
        finally:
            self._lock.release()
    
    def _reduce_finished(self, metadata):
        log.debug(u'reduce_finished(metadata=%s)' % (str(metadata)))
        self._runningReduces -= 1
        self._reduceQueue.put(metadata)
        self._dispatch_reduce()
        self._check_finished()
    
    def _reduce_failure(self, failure):
        self._runningReduces -= 1
        log.error('reduce_failure(failure=%s)' % (str(failure)))
        self._errors = True
        self.errback(failure)

def mapreduce(*args, **kwargs):
    return MapReduceDeferred(*args, **kwargs)